package reader

import (
	"encoding/json"
	"go.uber.org/zap"
	"strings"

	"v8s-ssp-proxy/internal/types"

	"github.com/Shopify/sarama"
)

type consumer struct {
	log *zap.SugaredLogger

	messages chan types.SSPPartnerMessage
	ready    chan bool
}

func (c *consumer) Setup(session sarama.ConsumerGroupSession) error {
	close(c.ready)
	return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (c *consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}

func (c *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for rawMessage := range claim.Messages() {

		c.log.With("message", string(rawMessage.Value)).Infof("received message from %s topic", rawMessage.Topic)
		var msg types.SSPPartnerMessage

		for _, h := range rawMessage.Headers {
			// if strings.Contains(string(h.Value), "billing") {

			// }
			msg.Status = strings.Contains(string(h.Value), "resource_status")
		}

		err := json.Unmarshal(rawMessage.Value, &msg)
		if err != nil {
			c.log.With("message", rawMessage.Value).Errorf("error unmarshal message from queue: %s", err)
			continue
		}

		c.messages <- msg

		session.MarkMessage(rawMessage, "")
	}

	return nil
}
